package com.dabai.iot.mqtt.server;

import io.vertx.core.Future;
import io.vertx.core.VerticleBase;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

/**
 * VertxMqttServer
 *
 * @author kev1n
 */
@Slf4j
public class VertxMqttServer extends VerticleBase {

    @Resource
    private MqttConnectionHandler mqttConnectionHandler;

    private MqttServer mqttServer;

    @Override
    public Future<?> start() {
        MqttServerOptions options = new MqttServerOptions()
            .setMaxMessageSize(204800)
            .setPort(1883)
            .setHost("0.0.0.0");

        mqttServer = MqttServer.create(vertx, options);

        // 先设置 handler
        mqttServer
            .exceptionHandler(error -> {
                log.error(error.getMessage(), error);
            })
            .endpointHandler(endpoint -> {

                if (mqttConnectionHandler.connectHandler(endpoint)) {
                    mqttConnectionHandler.publishHandler(endpoint);
                    mqttConnectionHandler.subscribeHandler(endpoint);
                    mqttConnectionHandler.unsubscribeHandler(endpoint);
                    mqttConnectionHandler.closeHandler(endpoint);

                    endpoint.pingHandler(v -> log.info("MQTT Server ====> Ping received from client: {}, remoteAddress: {}", endpoint.clientIdentifier(), endpoint.remoteAddress()));
                }
            });

        return mqttServer.listen().onComplete(ar -> {
            if (ar.succeeded()) {
                log.info("MQTT Server ====> Server started and listening on port {}, deploymentId {}", mqttServer.actualPort(), this.deploymentID());
            } else {
                log.info("MQTT Server ====> Server error on start，{}", ar.cause().getMessage());
            }
        });
    }

    @Override
    public Future<?> stop() {
        log.info("MQTT Server ====> Server closing, deploymentId {}", this.deploymentID());

        if (mqttServer != null) {
            return mqttServer.close().onComplete(ar -> {
                if (ar.succeeded()) {
                    log.info("MQTT Server ====> Server closed successfully");
                } else {
                    log.error("MQTT Server ====> Error closing MQTT server: {}", ar.cause().getMessage(), ar.cause());
                }
            });
        }

        return Future.succeededFuture();
    }

}
